package com.killer.mq;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.killer.mq.Constant;
import com.killer.pojo.entity.BOrder;
import com.killer.pojo.entity.Message;
import com.killer.service.BOrderService;
import com.killer.service.FileService;
import com.killer.service.MessageService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;

@Slf4j
@Component
@RocketMQMessageListener(topic = Constant.topic,consumerGroup = "order-consumer-group")
public class MessageConsumer implements RocketMQListener<Message> {


    @Autowired
    private MessageService messageService;


    @Autowired
    private FileService fileService;


    @Autowired
    private BOrderService orderService;


    @Override
    public void onMessage(Message message) {
        log.info("消费者接受到消息：" + message.toString());
        handleMessage(message);
    }

    //消息处理
    //todo 入参改成id 不应该用数据库对象
    private void handleMessage(Message message){
        //如果消息重复，根据版本号和id号进行过滤
        log.info("开始消费。。。。");
        // 更新状态 将版本号+1
        // TODO 先查一下状态
        int curState = getCurVersion((message.getId()));
        boolean flag = updateMessage(message,curState);
        //更新失败则说明是重复数据不消费
        if(flag == false) return;
        //todo 不是一次性读
        fileService.export(message);
    }


    public Integer getCurVersion(Long id){
       Message message = messageService.getById(id);
       return message.getVersion();
    }

    private boolean updateMessage(Message message,int curState) {
        UpdateWrapper<Message> updateWrapper = new UpdateWrapper<>();
        //version换成状态机
        updateWrapper.eq("state", curState);
        updateWrapper.eq("id", message.getId());
        Message msg = new Message();
        msg.setIsConsumer(1);
        //版本号加一
        msg.setVersion(curState + 1);
        boolean flag = messageService.update(msg,updateWrapper);
        return flag;
    }


}
